package com.persagy.iot.func

import com.persagy.iot.bean.IOTData
import com.persagy.iot.utils.HbaseUtil
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.hadoop.hbase.TableName
import org.apache.hadoop.hbase.client.{Connection, Put, Table}
import org.apache.hadoop.hbase.util.Bytes

class IOTAlarmSink extends RichSinkFunction[IOTData] {


  val original_tableName: String = "alarm_original_data"
  val family: String = "data"
  val sysTime: String = "sys_time"
  val status: String = "status"
  val value: String = "value"

  var conn: Connection = _
  var originalTable: Table = _
  override def open(parameters: Configuration): Unit = {
    println("init Hbase connection ......")
    conn = HbaseUtil.getHbaseConnection()

    println("create Hbase table ......")
    originalTable = conn.getTable(TableName.valueOf(original_tableName))
  }

  override def invoke(iotData: IOTData, context: SinkFunction.Context[_]): Unit = {
    // rowKey 按照 楼号:表号:功能号:时间
    val rowKey: String = iotData.rowKey + ":" + iotData.eventTime
    val put = new Put(Bytes.toBytes(rowKey))

    put.addColumn(Bytes.toBytes(family), Bytes.toBytes(sysTime), Bytes.toBytes(iotData.sysTime))
    originalTable.put(put)
    println("insert into " + original_tableName + " rowKey = " + rowKey)
  }

  override def close(): Unit = {
    try {
      if (originalTable != null) {
        originalTable.close()
      }
      if (conn != null) {
        conn.close()
      }
      println("init Hbase create connection ")
    } catch {
      case e:Exception => System.err.println(e.getMessage)
    }
  }
}
